package cn.edu.cug.cs.gtl.docsrv.doc;

import cn.edu.cug.cs.gtl.common.Pair;
import cn.edu.cug.cs.gtl.mybatis.Session;
import cn.edu.cug.cs.gtl.mybatis.mapper.doc.*;
import cn.edu.cug.cs.gtl.mybatis.metadata.DocUtils;
import cn.edu.cug.cs.gtl.protos.TimeValue;

import java.io.*;
import java.math.BigDecimal;
import java.net.URI;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import cn.edu.cug.cs.gtl.protoswrapper.TimeValueWrapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;



/**
 * 基于Oracle+HDFS实现的文档数据管理器
 * DOC_ITEM表中存放文档数据的条目信息
 * HDFS中存放文档文件
 * 通过DOC_ITEM表中的DOC_URL指向服务器中的文件全名
 */
public class HDFSDocManager extends DocManager{

    private Configuration conf; //hadoop 配置信息，使用 datanode hostname
    private String url;//hdfs url      hdfs://202.114.194.1:9000


    public HDFSDocManager(Session s) {
        super(s);
        this.conf = new Configuration();
        this.url = "hdfs://202.114.194.1:9000";
        conf.set("dfs.client.use.datanode.hostname","true");
    }


    private String getHdfsRemotePath(String docUrl){
        String[] ss = docUrl.split("/");
        StringBuilder stringBuilder = new StringBuilder();
        for (int i = 3; i < ss.length; i++) {
            stringBuilder.append("/");
            stringBuilder.append(ss[i]);
        }
        String hdfsPath = stringBuilder.toString();
        return hdfsPath;
    }

    /**
     *
     * @param docItem
     * @param docFile
     * @return
     */
    @Override
    public boolean insert(DocItem docItem, DocFile docFile){
        byte[] bytes = docFile.getDocRawBytes();
        String docUrl = docItem.getDocUrl();
        String hdfsPath = getHdfsRemotePath(docUrl);
        try{
            FileSystem fs = FileSystem.get(URI.create(url),conf);
            ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
            FSDataOutputStream outputStream = fs.create(new Path(hdfsPath));
            IOUtils.copyBytes(inputStream,outputStream,conf);
            fs.close();
        }catch (IOException e){
            e.printStackTrace();
        }
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        if(docItem.getDocInternalId()==null){
            docItem.setDocInternalId(docItemMapper.nextDocInternalId());
        }
        docItemMapper.insert(docItem);
        session.commit();
        return true;
    }


    /**
     *
     * @param docAll
     * @return
     */
    @Override
    public  boolean insert(DocAll docAll){
        Pair<DocItem,DocFile> pair = DocUtils.fromDocAll(docAll);
        return insert(pair.getFirst(), pair.getSecond());
    }


    /**
     *
     * @param docItem 必须填充docInternalId
     * @return
     */
    @Override
    public boolean delete(DocItem docItem) {
        //解析docUrl 在ftp上，docUrl应该是这种形式 "ftp://ip:port/remoteDir1/remoteDir2/fileName"
        String docUrl = docItem.getDocUrl();
        String hdfsPath = getHdfsRemotePath(docUrl);
        //delete the file on hdfs
        try{
            FileSystem fs = FileSystem.get(URI.create(url),conf);
            fs.deleteOnExit(new Path(hdfsPath));
        }catch (IOException e) {
            e.printStackTrace();
        }
        //delete the docItem
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        docItemMapper.deleteByPrimaryKey(docItem.getDocInternalId());
        session.commit();
        return true;

    }

    /**
     *
     * @param docAll 必须填充docInternalId
     * @return
     */
    @Override
    public boolean delete(DocAll docAll) {
        return delete(DocUtils.fromDocAll(docAll).getFirst());
    }


    @Override
    public boolean update(DocItem docItem, DocFile docFile) {
        //update docItem
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        docItemMapper.updateByPrimaryKey(docItem);
        // file on hdfs is not allowed to update

        session.commit();
        return true;
    }

    @Override
    public boolean update(DocAll docAll) {
        return update(DocUtils.fromDocAll(docAll).getFirst(),DocUtils.fromDocAll(docAll).getSecond());
    }

    @Override
    public DocAll queryByInternalId(BigDecimal docInternalId) {
        //从数据库中提取条目信息
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        DocItem docItem =  docItemMapper.selectByPrimaryKey(docInternalId);
        //hdfs url:   hdfs://202.114.194.1:9000/remoteDir/aaa/remoteFileName
        //解析hdfs 地址
        String docUrl = docItem.getDocUrl();
        String hdfsPath = getHdfsRemotePath(docUrl);
        byte[] bytes = null;
        try{
            FileSystem fs = FileSystem.get(URI.create(url),conf);
            // get the bytes from hdfs
            FSDataInputStream inputStream = fs.open(new Path(hdfsPath));
            bytes = new byte[inputStream.available()];
            inputStream.readFully(bytes);
            inputStream.close();
            fs.close();
        }catch (IOException e){
            e.printStackTrace();
        }
        DocFile docFile = new DocFile();
        docFile.setDocInternalId(docItem.getDocInternalId());
        docFile.setDocRawBytes(bytes);
        DocAll docAll = DocUtils.toDocAll(docItem,docFile);
        session.commit();
        return docAll;

    }

    @Override
    public List<DocAll> queryByTitle(String docEnTitle) {
        List<DocAll> res = new ArrayList<>();
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        List<DocItem> docItems = docItemMapper.selectByEnTitle(docEnTitle);
        for (DocItem docItem : docItems){
            String docUrl = docItem.getDocUrl();
            String hdfsPath = getHdfsRemotePath(docUrl);
            byte[] bytes = null;
            try{
                FileSystem fs = FileSystem.get(URI.create(url),conf);
                FSDataInputStream inputStream = fs.open(new Path(hdfsPath));
                bytes = new byte[inputStream.available()];
                inputStream.readFully(bytes);
                inputStream.close();
                fs.close();
            }catch (IOException e){
                e.printStackTrace();
            }
            DocFile docFile = new DocFile();
            docFile.setDocInternalId(docItem.getDocInternalId());
            docFile.setDocRawBytes(bytes);
            DocAll docAll = DocUtils.toDocAll(docItem,docFile);
            res.add(docAll);
        }
        session.commit();
        return res;
    }

    @Override
    public List<DocAll> queryByComment(String docCnTitle) {
        List<DocAll> res = new ArrayList<>();
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        List<DocItem> docItems = docItemMapper.selectByCnTitle(docCnTitle);
        for (DocItem docItem : docItems){
            String docUrl = docItem.getDocUrl();
            String hdfsPath = getHdfsRemotePath(docUrl);
            byte[] bytes = null;
            try{
                FileSystem fs = FileSystem.get(URI.create(url),conf);
                FSDataInputStream inputStream = fs.open(new Path(hdfsPath));
                bytes = new byte[inputStream.available()];
                inputStream.readFully(bytes);
                inputStream.close();
                fs.close();
            }catch (IOException e){
                e.printStackTrace();
            }
            DocFile docFile = new DocFile();
            docFile.setDocInternalId(docItem.getDocInternalId());
            docFile.setDocRawBytes(bytes);
            DocAll docAll = DocUtils.toDocAll(docItem,docFile);
            res.add(docAll);
        }
        session.commit();
        return res;
    }

    @Override
    public List<DocAll> queryByType(String docType) {
        List<DocAll> res = new ArrayList<>();
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        List<DocItem> docItems = docItemMapper.selectByType(docType);
        for (DocItem docItem : docItems){
            String docUrl = docItem.getDocUrl();
            String hdfsPath = getHdfsRemotePath(docUrl);
            byte[] bytes = null;
            try{
                FileSystem fs = FileSystem.get(URI.create(url),conf);
                FSDataInputStream inputStream = fs.open(new Path(hdfsPath));
                bytes = new byte[inputStream.available()];
                inputStream.readFully(bytes);
                inputStream.close();
                fs.close();
            }catch (IOException e){
                e.printStackTrace();
            }
            DocFile docFile = new DocFile();
            docFile.setDocInternalId(docItem.getDocInternalId());
            docFile.setDocRawBytes(bytes);
            DocAll docAll = DocUtils.toDocAll(docItem,docFile);
            res.add(docAll);
        }
        session.commit();
        return res;
    }

    @Override
    public List<DocAll> queryByAppendDate(TimeValue begin, TimeValue end) {
        List<DocAll> res = new ArrayList<>();
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        List<DocItem> docItems = docItemMapper.selectByAppendDate(TimeValueWrapper.toUtilDate(begin),TimeValueWrapper.toUtilDate(end));
        for (DocItem docItem : docItems){
            String docUrl = docItem.getDocUrl();
            String hdfsPath = getHdfsRemotePath(docUrl);
            byte[] bytes = null;
            try{
                FileSystem fs = FileSystem.get(URI.create(url),conf);
                FSDataInputStream inputStream = fs.open(new Path(hdfsPath));
                bytes = new byte[inputStream.available()];
                inputStream.readFully(bytes);
                inputStream.close();
                fs.close();
            }catch (IOException e){
                e.printStackTrace();
            }
            DocFile docFile = new DocFile();
            docFile.setDocInternalId(docItem.getDocInternalId());
            docFile.setDocRawBytes(bytes);
            DocAll docAll = DocUtils.toDocAll(docItem,docFile);
            res.add(docAll);
        }
        session.commit();
        return res;
    }




    @Override
    public boolean insert(DocItem docItem, String localSingleFile) {
        try{
            String docUrl = docItem.getDocUrl();
            String hdfsPath = getHdfsRemotePath(docUrl);
            FileSystem fs = FileSystem.get(URI.create(url),conf);
            fs.copyFromLocalFile(new Path(localSingleFile),new Path(hdfsPath));
        }catch (IOException e){
            e.printStackTrace();
        }
        DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
        if(docItem.getDocInternalId()==null) {
            docItem.setDocInternalId(docItemMapper.nextDocInternalId());
        }
        docItemMapper.insert(docItem);
        session.commit();
        return true;
    }

    @Override
    public boolean insert(String localDir) {
        try{
            FileSystem fs = FileSystem.get(URI.create(url),conf);
            File file = new File(localDir);
            File[] files = file.listFiles();
            for (File f : files){
                String hdfsPath = f.getName();
                InputStream inputStream = new FileInputStream(f);
                OutputStream outputStream = fs.create(new Path(hdfsPath));
                IOUtils.copyBytes(inputStream,outputStream,conf);

                DocItemMapper docItemMapper = session.getMapper(DocItemMapper.class);
                DocItem docItem = new DocItem();
                docItem.setDocAppendDate(new Date());
                String[] ss = localDir.split("/");
                docItem.setDocUrl("hdfs://202.114.194.1:9000/"+ ss[ss.length-1]+ "/" + f.getName());
                docItem.setDocEnTitle(f.getName());
                docItem.setDocInternalId(docItemMapper.nextDocInternalId());
                docItemMapper.insert(docItem);
            }
            fs.close();
        }catch (IOException e ){
            e.printStackTrace();
        }
        session.commit();
        return true;
    }





}
